package org.apache.zookeeper.server.paxos2;

import java.util.*;

import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.proto.*;
import org.apache.zookeeper.server.ZooKeeperServer;
import org.apache.zookeeper.txn.*;
import org.apache.log4j.Logger;

public class Mapping {
    private static final Logger LOG = Logger.getLogger(Mapping.class);

    public static int keyToShard(long key) {
        //return (int)(key*Configuration.blockSize / (Configuration.totalSize/Configuration.noOfStripes)) +1;
        return (int) (key % Configuration.noOfStripes) + 1;
    }

    public static long keyInShard(long key) {
        //return (int)(key*Configuration.blockSize / (Configuration.totalSize/Configuration.noOfStripes)) +1;
        return key / Configuration.noOfStripes;
    }


    public static long keyToOffsetInShard(long key) {
        if (Configuration.writeQuorumSize < Configuration.noOfStripes)
            return key / Configuration.noOfStripes * Configuration.blockSize + Configuration.keyMapping[Configuration.serverId - 1][(int) (key % Configuration.noOfStripes)] * (Configuration.totalSize / Configuration.noOfStripes);
        else
            return key * Configuration.blockSize;
    }

    public static long getBlock(int shardId, long keyInShard) {
        return keyInShard * Configuration.noOfStripes + (shardId - 1);
    }

    public static List<Integer> listShards(int nodeId) {
        LinkedList<Integer> ret = new LinkedList<Integer>();
        for (int i = 0; i < Configuration.writeQuorumSize; i++) {
            int tmp = (nodeId + i) % (int) Configuration.noOfStripes;
            if (tmp == 0)
                ret.add((int) Configuration.noOfStripes);
            else
                ret.add(tmp);
        }
        return ret;
    }

    public static List<Integer> shardToNode(int stripeId, List<Integer> nodes) {
        LinkedList<Integer> ret = new LinkedList<Integer>();
        for (int i = 0; i < Configuration.writeQuorumSize; i++) {
            if (stripeId - i > 0)
                if (nodes != null)
                    ret.add(nodes.get((stripeId - i - 1) % nodes.size()));
                else
                    ret.add(stripeId - i);
            else if (nodes != null)
                ret.add(nodes.get(stripeId - i + nodes.size() - 1) % nodes.size());
            else
                ret.add((int) (stripeId - i + Configuration.noOfStripes));
        }
        return ret;
    }

    public static List<Integer> keyToNode(long key, List<Integer> nodes) {
        return shardToNode(keyToShard(key), nodes);
    }

    public static long getBlockId(Request request) {
        try {
            if (request.type == OpCode.getData) {
                request.request.rewind();
                GetDataRequest getDataRequest = new GetDataRequest();
                ZooKeeperServer.byteBuffer2Record(request.request,
                        getDataRequest);
                return getDataRequest.getBlockId();
            } else if (request.type == OpCode.setData) {
                if (request.txn != null)
                    return ((SetDataTxn) (request.txn)).getBlockId();
                request.request.rewind();
                SetDataRequest setDataRequest = new SetDataRequest();
                ZooKeeperServer.byteBuffer2Record(request.request,
                        setDataRequest);
                return setDataRequest.getBlockId();
            } else if (request.type == OpCode.boloskyGetData) {
                request.request.rewind();
                BoloskyGetDataRequest getDataRequest = new BoloskyGetDataRequest();
                ZooKeeperServer.byteBuffer2Record(request.request,
                        getDataRequest);
                return getDataRequest.getBlockId();
            } else
                throw new Exception("GetBlockId for a non read-write operation");
        }
        catch (Exception e) {
            LOG.error("Error:: ", e);
        }
        return -1;
    }

    /*public static boolean isMyRequest(Request request) {
        if (request.type != OpCode.getData && request.type != OpCode.setData)
            return true;

	 try {
            if (request.type == OpCode.getData) {
                return true;
            } else if (request.type == OpCode.setData) {
                request.request.rewind();
                SetDataRequest setDataRequest = new SetDataRequest();
                ZooKeeperServer.byteBuffer2Record(request.request,
                        setDataRequest);
		byte[] quorum = setDataRequest.getQuorum();
		for(int i=0; i<quorum.length; i++)
		     if(Configuration.serverId==(int)quorum[i])
			return true;
            } else
                throw new Exception("GetBlockId for a non read-write operation");
        }
        catch (Exception e) {
            LOG.error("Error:: ", e);
        }
	return false;


    }*/


}
